package cn.com.duiba.millionaire.center.biz.service.impl;

import cn.com.duiba.millionaire.center.api.constant.RedisKeyFactory;
import cn.com.duiba.millionaire.center.biz.service.QuestionSelectionStastisticService;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Created by zzy on 2018/1/20.
 */
@Service
public class QuestionSelectionStastisticServiceImpl implements QuestionSelectionStastisticService {

    @Resource(name = "redisTemplate")
    private RedisTemplate<String, String> redisTemplate;

    @Override
    public void batchAdd(Map<Long, Map<Integer, Map<String, Integer>>> batchData) {
        //key1:K001_活动id_第几题,key2:userAnswer(用户选择的答案)，value2:选择该答案的人数
        Map<String, Map<String, Integer>> mappedData = new HashMap<>();
        for (Map.Entry<Long, Map<Integer, Map<String, Integer>>> entry : batchData.entrySet()) {
            for (Map.Entry<Integer, Map<String, Integer>> entryInner :
                    entry.getValue().entrySet()) {
                mappedData.put(buildKey(entry.getKey(), entryInner.getKey()), entryInner.getValue());
            }
        }

        //更新redis中每个活动下每一题选择人数
        redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer<Object> keyS = (RedisSerializer<Object>) redisTemplate.getKeySerializer();
                RedisSerializer<Object> valueS = (RedisSerializer<Object>) redisTemplate.getValueSerializer();
                for (Map.Entry<String, Map<String, Integer>> entry : mappedData.entrySet()) {
                    String key = entry.getKey();
                    byte[] keyByte = keyS.serialize(key);
                    for (Map.Entry<String, Integer> entryInner : entry.getValue().entrySet()) {
                        connection.zIncrBy(keyByte, entryInner.getValue().doubleValue(), valueS.serialize(entryInner.getKey()));
                    }
                    // 保存3天
                    connection.expire(keyByte, 3 * 24 * 3600);
                }
                return null;
            }
        });
    }

    @Override
    public Map<String, Long> findQuestionStastistic(Long activityId, Integer questionIndex) {
        String key = buildKey(activityId, questionIndex);
        Set<ZSetOperations.TypedTuple<String>> typedTupleSet = redisTemplate.opsForZSet().reverseRangeWithScores(key, 0, Integer.MAX_VALUE);
        if (CollectionUtils.isEmpty(typedTupleSet)) {
            return Collections.emptyMap();
        }
        Map<String, Long> map = new HashMap<>();
        typedTupleSet.forEach(tup -> {
            map.put(tup.getValue(), tup.getScore().longValue());
        });
        return map;
    }

    private String buildKey(Long activityId, Integer questionIndex) {
        return RedisKeyFactory.K001.toString() + activityId + "_" + questionIndex;
    }
}
